package com.tvunetworks.center.device.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.tvunetworks.center.common.model.vo.TimeLockDeviceMsgVo;
import com.tvunetworks.center.common.util.JsonUtil;
import com.tvunetworks.center.common.util.MyUtil;
import com.tvunetworks.center.common.util.Result;
import com.tvunetworks.center.device.constant.CommandConstant;
import com.tvunetworks.center.device.constant.DeviceConstant;
import com.tvunetworks.center.device.mapper.TimeLockMsgMapper;
import com.tvunetworks.center.device.model.TimeLockEvent;
import com.tvunetworks.center.device.model.TimeLockMsg;
import com.tvunetworks.center.device.model.dto.TimeLockBody;
import com.tvunetworks.center.device.model.vo.TimeLockDeviceVo;
import com.tvunetworks.center.device.model.vo.TimeLockMsgVO;
import com.tvunetworks.center.device.service.MsgpipeService;
import com.tvunetworks.center.device.service.ReceiverControlService;
import com.tvunetworks.center.device.service.TimeLockEventService;
import com.tvunetworks.center.device.service.TimeLockMsgService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @program: tvucc-aggregator
 * @description: TimeLockMsgServiceImpl
 * @author lebronchen
 * @create: 2019-03-26 10:32
 **/
@Service
@Slf4j
public class TimeLockMsgServiceImpl extends BaseService<TimeLockMsg> implements TimeLockMsgService {

    @Autowired
    private MsgpipeService msgpipeService;
    @Autowired
    private TimeLockMsgMapper timeLockMsgMapper;
    @Autowired
    private TimeLockEventService timeLockEventService;
    @Autowired
    private ReceiverControlService receiverControlService;

    @Override
    public void reportTimeLockMsg(String eventId, String mainPeerId, List<String> synPeerIdList, Map<String, String> peerIdMap,
                                  Map<String, String> packDependenceRidMap) {
        TimeLockMsg timeLockMsg = new TimeLockMsg()
                .setEventId(eventId)
                .setPeerId(mainPeerId)
                .setMsgType(DeviceConstant.TimeLockEventType.REPORT_SYNC)
                .setMainServer(1)
                .setStatus(1)
                .setIntranetIp(peerIdMap.get(mainPeerId));
        super.insert(timeLockMsg);
        this.reportSyncDeviceTimeLockMsg(synPeerIdList, peerIdMap, timeLockMsg, packDependenceRidMap);
    }

    @Override
    public void reportSyncDeviceTimeLockMsg(List<String> synPeerIdList, Map<String, String> peerIdMap, TimeLockMsg timeLockMsg,
                                            Map<String, String> packDependenceRidMap) {
        List<String> sendMsgPipePeerIds = this.checkSendMsgPipe(peerIdMap, timeLockMsg.getEventId(), timeLockMsg.getPeerId(), packDependenceRidMap);
        String mainPeerId = timeLockMsg.getPeerId();
        if (!CollectionUtils.isEmpty(synPeerIdList)) {
            for (String peerId : synPeerIdList) {
                timeLockMsg.setId(null);
                timeLockMsg.setPeerId(peerId);
                timeLockMsg.setMainServer(0);
                timeLockMsg.setStatus(0)
                        .setLiveRid(packDependenceRidMap.get(peerId))
                        .setIntranetIp(peerIdMap.get(peerId))
                        .setReceiverStatus(null);
                if (!sendMsgPipePeerIds.contains(peerId.toLowerCase())) {
                    timeLockMsg.setReceiverStatus("-1");
                    if (StringUtils.equalsIgnoreCase(peerIdMap.get(peerId), peerIdMap.get(mainPeerId))) {
                        timeLockMsg.setStatus(1);
                    }
                    log.error("reportSyncDeviceTimeLockMsg>>>>>>not send message to msgPipe.peerId:[{}],timeLockMsg:[{}] ", peerId, timeLockMsg);
                }
                super.insert(timeLockMsg);
            }
        }
    }

    /**
     * 校验是否发送给msgpipe
     * @param peerIdMap key为peerId，value为intranetIp
     * @param eventId  timeLock的eventId
     * @param mainPeerId 主R的PeerId
     * @param packDependenceRidMap pack对应得rid
     * @return 返回需要下发给msgPipe的peerId集合
     */
    private List<String> checkSendMsgPipe(Map<String, String> peerIdMap, String eventId,String mainPeerId,
                                          Map<String, String> packDependenceRidMap) {
        List<String> sendMsgPipePeerIds = new ArrayList<>();
        if (packDependenceRidMap != null && packDependenceRidMap.size() > 0) {
            sendMsgPipePeerIds.addAll(packDependenceRidMap.keySet().stream().map(String::toLowerCase).collect(Collectors.toSet()));
        }
        if (peerIdMap == null || peerIdMap.size() == 0) {
            return sendMsgPipePeerIds;
        }
        //获取已经存在的timeLockMsg的intranetIp集合
        Set<String> intranetIps = new HashSet<>();
        List<TimeLockMsg> timeLockMsgs = this.listSynchronizingMsgByEventId(eventId);
        if (!CollectionUtils.isEmpty(timeLockMsgs)) {
            Set<String> collect = timeLockMsgs.stream().map(TimeLockMsg::getIntranetIp).collect(Collectors.toSet());
            if (!CollectionUtils.isEmpty(collect)) {
                intranetIps.addAll(collect);
            }
        }
        if (StringUtils.isNotEmpty(mainPeerId)) {
            String mainIntranetIp = peerIdMap.get(mainPeerId);
            if (StringUtils.isNotEmpty(mainIntranetIp)) {
                intranetIps.add(mainIntranetIp);
            }
        }
        for (Map.Entry<String, String> stringStringEntry : peerIdMap.entrySet()) {
            String peerId = stringStringEntry.getKey();
            if (StringUtils.equalsIgnoreCase(peerId, mainPeerId)) {
                continue;
            }
            String intranetIp = stringStringEntry.getValue();
            if (StringUtils.isEmpty(intranetIp)) {
                log.error("checkSendMsgPipe>>>>>>>>>intranetIp is empty,peerId:[{}]", peerId);
                sendMsgPipePeerIds.add(peerId.toLowerCase());
                continue;
            }
            if (intranetIps.contains(intranetIp)) {
                log.error("checkSendMsgPipe>>>>>>intranetIp already exist,peerId:[{}],intranetIp:[{}]", peerId, intranetIp);
            }else {
                sendMsgPipePeerIds.add(peerId.toLowerCase());
                intranetIps.add(intranetIp);
            }
        }
        return sendMsgPipePeerIds;
    }

    @Override
    public void startTimeLockMsg(String eventId, String mainPeerId, List<String> synPeerIdList, TimeLockMsgVO msgVO,
                                 Map<String, String> peerIdMap, Map<String, String> packDependenceRidMap) {
        msgVO.setEventId(eventId);
        msgVO.setMsgType(DeviceConstant.TimeLockEventType.START_SYNC);
        msgVO.setMainServer(true);
        String msg = null;
        try {
            msg = MyUtil.MAPPER.writeValueAsString(msgVO);
        } catch (JsonProcessingException e) {
            log.error("startTimeLockMsg>>>>>>main Receiver parse json exception,msgVO:[{}],e:[{}]", msgVO, e);
        }
        TimeLockMsg timeLockMsg = new TimeLockMsg();
        timeLockMsg.setEventId(eventId);
        timeLockMsg.setPeerId(mainPeerId);
        timeLockMsg.setMessage(msg);
        timeLockMsg.setMainServer(1);
        timeLockMsg.setIntranetIp(peerIdMap.get(mainPeerId));
        timeLockMsg.setMsgType(DeviceConstant.TimeLockEventType.START_SYNC);
        msgpipeService.sendMsgToMsgpipe(mainPeerId, msg);
        super.insert(timeLockMsg);
        this.startSyncDeviceTimeLockMsg(synPeerIdList, msgVO, peerIdMap, timeLockMsg, packDependenceRidMap);
    }

    @Override
    public void startSyncDeviceTimeLockMsg(List<String> synPeerIdList, TimeLockMsgVO msgVO, Map<String, String> peerIdMap,
                                           TimeLockMsg timeLockMsg, Map<String, String> packDependenceRidMap) {
        List<String> sendMsgPipePeerIds = this.checkSendMsgPipe(peerIdMap, timeLockMsg.getEventId(), timeLockMsg.getPeerId(), packDependenceRidMap);
        if (synPeerIdList != null && synPeerIdList.size() > 0) {
            msgVO.setMainServer(false);
            String msg = null;
            try {
                msg = MyUtil.MAPPER.writeValueAsString(msgVO);
            } catch (JsonProcessingException e) {
                log.error("startTimeLockMsg>>>>>> receiver/pack parse json exception,msgVO:[{}],e:[{}]", msgVO, e);
            }
            for (String peerId : synPeerIdList) {
                timeLockMsg.setId(null);
                timeLockMsg.setPeerId(peerId);
                timeLockMsg.setMessage(msg);
                timeLockMsg.setMsgType(DeviceConstant.TimeLockEventType.START_SYNC);
                timeLockMsg.setMainServer(0);
                timeLockMsg.setIntranetIp(peerIdMap.get(peerId));
                timeLockMsg.setLiveRid(packDependenceRidMap.get(peerId));
                if (sendMsgPipePeerIds.contains(peerId.toLowerCase())) {
                    timeLockMsg.setReceiverStatus(null);
                    msgpipeService.sendMsgToMsgpipe(peerId, msg);
                }else {
                    timeLockMsg.setReceiverStatus("-1");
                    log.error("startSyncDeviceTimeLockMsg>>>>>>>> intranetIp already send msgpipe,peerId:[{}],timeLockMsg:[{}]", peerId, timeLockMsg);
                }
                super.insert(timeLockMsg);
            }
        }
    }

    @Override
    public void stopTimeLockMsg(String eventId, String userId) {
        List<TimeLockMsg> timeLockMsgs = this.listSynchronizingPeerIdsByEventId(eventId);
        if (CollectionUtils.isEmpty(timeLockMsgs)) {
            return;
        }
        timeLockMsgs.forEach(timeLockMsg -> {
            stopTimeLockMsg(eventId, timeLockMsg.getPeerId(), Boolean.TRUE, userId,
                    !StringUtils.equalsIgnoreCase(timeLockMsg.getReceiverStatus(), "-1"));
        });
    }

    @Override
    @Transactional
    public void stopTimeLockMsg(String eventId, String peerId, boolean needSendMsgpipe, String userId, boolean sendMsgPipe) {
        TimeLockMsgVO msgVO = new TimeLockMsgVO();
        msgVO.setEventId(eventId);
        msgVO.setMsgType(DeviceConstant.TimeLockEventType.STOP_SYNC);
        String msg = null;
        try {
            msg = JsonUtil.toJsonStr(msgVO);
        } catch (IOException e) {
            log.error(e.toString(), e);
        }
        if (needSendMsgpipe) {
            // 添加Msg记录到DB
            TimeLockMsg timeLockMsg = new TimeLockMsg();
            timeLockMsg.setOperationUserId(userId);
            if (sendMsgPipe) {
                // 发送消息到msgpipe
                msgpipeService.sendMsgToMsgpipe(peerId, msg);
                //发送消息,让R端停止TimeLock,这是发给msgpipe,misgpipe收到消息后,则将ReceiverStatus状态置为回信成功(1),但是不确定是否R端是否消费.
                // 有可能其他服务挂了导致R端没收到,将会有定时任务checkR是否消费.如果没有,则定时任务执行关闭操作.
                try {
                    //获取R的小版本信息,因为定时任务的check接口,只支持7730版本以上的R
                    Result<String> versionR = receiverControlService.getVersionR(peerId, CommandConstant.GET_VERSION_R);
                    if (versionR.getResult() != null) {
                        Map<String, Object> result = JsonUtil.toObject(versionR.getResult(), Map.class);
                        String version = (String) result.get("StrVersion");
                        String[] versions = StringUtils.split(version, "build");
                        String buildVersion = versions[1];
                        if (Integer.parseInt(buildVersion) > 7730) {
                            timeLockMsg.setReceiverStatus("1");
                        }
                    }
                } catch (Exception e) {
                    log.error("get  Receiver Version --->>>>peerId:[{}]", peerId, e);
                }
            }
            timeLockMsg.setEventId(eventId);
            timeLockMsg.setMsgType(msg);
            timeLockMsg.setPeerId(peerId);
            timeLockMsg.setMsgType(DeviceConstant.TimeLockEventType.STOP_SYNC);
            super.insert(timeLockMsg);
        }
    }

    @Override
    public List<TimeLockMsg> listSynchronizingPeerIdsByEventId(String eventId) {
        TimeLockMsg timeLockMsg = new TimeLockMsg();
        timeLockMsg.setEventId(eventId);
        timeLockMsg.setMsgType(DeviceConstant.TimeLockEventType.START_SYNC);
        return super.selectList(timeLockMsg);
    }

    @Override
    public List<TimeLockMsg> listSynchronizingMsgByEventId(String eventId) {
        return timeLockMsgMapper.listSynchronizingMsgByEventId(eventId);
    }

    @Override
    public List<TimeLockDeviceVo> listSynchronizingStartReceiverMsgByEventId(String eventId) {
        return timeLockMsgMapper.listSynchronizingStartReceiverMsgByEventId(eventId);
    }

    @Override
    public TimeLockMsg getTimeLockMsg(String eventId, String rid, String msgType) {
        TimeLockMsg timeLockMsg = new TimeLockMsg();
        timeLockMsg.setMsgType(msgType);
        timeLockMsg.setEventId(eventId);
        timeLockMsg.setPeerId(rid);
        return super.selectExistOne(timeLockMsg);
    }

    /**
     * 根据eventId和intranetIp获取所有的timeLockMsg信息
     * @param eventId
     * @param intranetIp
     * @param msgType
     * @return
     */
    private List<TimeLockMsg> getTimeLockMsgByEventIdAndIntranetIp(String eventId, String intranetIp, String msgType) {
        TimeLockMsg timeLockMsg = new TimeLockMsg();
        timeLockMsg.setMsgType(msgType);
        timeLockMsg.setEventId(eventId);
        timeLockMsg.setIntranetIp(intranetIp);
        return super.selectList(timeLockMsg);
    }


    @Override
    public List<TimeLockMsg> listTimeLockMsg(String eventId, List<String> peerIdList, String reportSync) {
        if (CollectionUtils.isEmpty(peerIdList)) {
            return null;
        }
        List<TimeLockMsg> resultList = new ArrayList<>();
        for (String peerId : peerIdList) {
            TimeLockMsg record = new TimeLockMsg()
                    .setEventId(eventId)
                    .setPeerId(peerId)
                    .setMsgType(reportSync);
            List<TimeLockMsg> list = super.selectExistList(record);
            if (!CollectionUtils.isEmpty(list)) {
                resultList.add(list.get(0));
            }
        }
        return resultList;
    }

    @Override
    public void reportSyncUpdateStatus(String peerId, TimeLockBody body) {
        TimeLockMsg exist = this.getTimeLockMsg(body.getEventId(), peerId, body.getMsgType());
        if (exist == null) {
            List<TimeLockMsg> timeLockMsgs = this.getTimeLockMsgByEventIdAndLiveRid(body.getEventId(), peerId, DeviceConstant.TimeLockEventType.START_SYNC);
            if (CollectionUtils.isEmpty(timeLockMsgs)) {
                log.error("reportSyncUpdateStatus>>>>>timeLockMsg is empty,peerId:[{}],body:[{}]", peerId, body);
                return;
            }
            TimeLockMsg record = new TimeLockMsg()
                    .setEventId(body.getEventId())
                    .setPeerId(peerId)
                    .setMsgType(body.getMsgType())
                    .setStatus(body.getStatus());
            super.insert(record);
        } else {
            if (body.getStatus() != exist.getStatus()) {
                exist.setStatus(body.getStatus());
                super.update(exist);
            }
            //将所有相同的intranetIp的状态也修改了
            if (StringUtils.isEmpty(exist.getIntranetIp())) {
                return;
            }
            List<TimeLockMsg> timeLockMsgs = this.getTimeLockMsgByEventIdAndIntranetIp(body.getEventId(), exist.getIntranetIp(), body.getMsgType());
            if (!CollectionUtils.isEmpty(timeLockMsgs)) {
                log.error("reportSyncUpdateStatus>>>>>>>> Same intranetIp of timeLockMsgs,intranetIp:[{}],timeLockMsgs:[{}]", exist.getIntranetIp(), timeLockMsgs);
                timeLockMsgs.forEach(timeLockMsg -> {
                    if (body.getStatus() != timeLockMsg.getStatus()) {
                        timeLockMsg.setStatus(body.getStatus());
                        super.update(timeLockMsg);
                    }
                });
            }
        }
    }

    @Override
    public int updateTimeLockMsgByEventIdAndPeerIdAndStatus(String peerId, String eventId, String msgType, int status) {
        try {
            Map<String, Object> map = new HashMap<>();
            map.put("eventId", eventId);
            map.put("msgType", msgType);
            map.put("status", status);
            String msgBody = MyUtil.MAPPER.writeValueAsString(map);
            TimeLockMsg example = new TimeLockMsg()
                    .setEventId(eventId)
                    .setPeerId(peerId)
                    .setMsgType(msgType);
            TimeLockMsg timeLockMsg = super.selectExistOne(example);
            if (timeLockMsg != null) {
                timeLockMsg.setStatus(status)
                        .setMessage(msgBody);
                super.updateSelective(timeLockMsg);
            }

        } catch (JsonProcessingException e) {
            log.error("updateTimeLockMsgByEventIdAndPeerIdAndStatus>>>>>>parseJson Exception:[{}]", e);
        }
        return 0;
    }

    @Override
    public int batchDeleteByPeerIdsAndEventId(List<String> peerIdList, String eventId) {
        if (CollectionUtils.isEmpty(peerIdList)) {
            return 0;
        }
        return timeLockMsgMapper.batchDeleteByPeerIdsAndEventId(peerIdList, eventId);
    }

    @Override
    public void logicDeleteMsgByEventIdAndPeerId(String eventId, String peerId) {
        timeLockMsgMapper.logicDeleteMsgByEventIdAndPeerId(eventId, peerId);
    }

    @Override
    public List<TimeLockDeviceMsgVo> listSynchronizingReportReceiverMsgByRids(List<String> rids) {
        return timeLockMsgMapper.listSynchronizingReportReceiverMsgByRids(rids);
    }

    @Override
    public List<TimeLockDeviceMsgVo> listOpenTimeLockPackByRids(List<String> rids) {
        return timeLockMsgMapper.listOpenTimeLockPackByRids(rids);
    }

    @Override
    public TimeLockEvent stopTimeLockMessageSendToReceiver(String eventId, String rid) {
        //1.根据eventId和rid获取timeLockEvent

        TimeLockEvent timeLockEvent = timeLockEventService.getTimeLockEventById(eventId);
        if (timeLockEvent == null) {
            log.error("stopTimeLockMessageSendToReceiver>>>>>>>>>>>>timeLockEvent is empty");
            return null;
        }

        TimeLockMsgVO msgVO = new TimeLockMsgVO();
        msgVO.setEventId(eventId);
        msgVO.setMsgType(DeviceConstant.TimeLockEventType.STOP_SYNC);
        String msg = null;
        try {
            msg = JsonUtil.toJsonStr(msgVO);
        } catch (IOException e) {
            log.error(e.toString(), e);
        }
        // 发送消息到msgpipe
        msgpipeService.sendMsgToMsgpipe(rid, msg);
        return timeLockEvent;
    }

    @Override
    public List<TimeLockMsg> getTimeLockMsgByEventIdAndLiveRid(String eventId, String rid, String msgType) {
        TimeLockMsg timeLockMsg = new TimeLockMsg();
        timeLockMsg.setMsgType(msgType);
        timeLockMsg.setEventId(eventId);
        timeLockMsg.setLiveRid(rid);
        return super.selectExistList(timeLockMsg);
    }
    /**
     * 检查R那边是否收到了停止TimeLock指令并且关闭成功
     * @return TimeLockMsg
     */
    @Override
    public  List<TimeLockMsg> checkReceiver() {
        List<TimeLockMsg> timeLockMsgs = timeLockMsgMapper.checkReceiver();
        return timeLockMsgs;
    }
    /**
     * 将已经正常关闭的R,DB中receiverStatus状态值设置为1,表示正常关闭.
     * @param id 主键Id
     */
    @Override
    public void updateReceiverTimeLockStatus(String id) {
        timeLockMsgMapper.updateReceiverTimeLockStatus(id);
    }
}
